package com.qingcloud.df.executor.consumer;

import com.qingcloud.df.executor.model.InstanceRuntimeExport;
import com.qingcloud.df.executor.constant.DateFormatterConstant;
import com.qingcloud.df.sdk.component.message.ComponentOutputLog;
import com.qingcloud.df.sdk.component.message.ComponentRunLog;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

import static cn.hutool.core.text.CharSequenceUtil.isBlank;
import static com.alibaba.fastjson.JSON.parseObject;

/**
 * 运行中心队列消费者
 *
 * @author wangjinnan
 */
@Component
@Slf4j
public class ExecutorConsumer {

    /**
     * 保存执行日志记录
     *
     * @param runLog 日志对象
     */
    protected void saveRuntimeLog(ComponentRunLog runLog) {
        //运行日志
        //TODO
//        this.instanceMapper.saveRuntimeLog(runLog);
    }

    /**
     * 保存组件输出结果
     *
     * @param outputLog 日志对象
     */
    protected void saveExport(ComponentOutputLog outputLog) {
        //组件输出
        try {
            final List<InstanceRuntimeExport> exportList = outputLog.getOutputList().stream().map(it -> {
                final InstanceRuntimeExport export = new InstanceRuntimeExport();
                export.setOutput(it);
                export.setInstanceId(Integer.parseInt(outputLog.getInstanceId()));
                export.setProduceTime(outputLog.getSendTime());
                export.setInsTime(LocalDateTime.now().format(DateFormatterConstant.NORMAL_FULL_FORMATTER));
                return export;
            }).collect(Collectors.toList());
            //TODO
//            this.instanceMapper.saveRuntimeExport(exportList);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    /**
     * 保存数据来源数据量
     */
    protected void saveOriginDataCount(ComponentOutputLog outputLog) {
        StringBuilder builder = new StringBuilder();
        for (String output : outputLog.getOutputList()) {
            builder.append(output);
        }
        //TODO
//        this.instanceMapper.updateDataOriginCount(instance.getDsId(), builder.length());
    }

    /**
     * 组件运行日志数据消费
     */
//    @KafkaListener(topics = ComponentRunLog.COMPONENT_LOG_QUEUE_TOPIC)
    public void consumerRunLog(List<String> dataList) {
        for (String data : dataList) {
            ComponentRunLog runLog = parseObject(data, ComponentRunLog.class);
            //保存条用记录
            saveRuntimeLog(runLog);
        }
    }

    /**
     * 组件运行日志数据消费
     */
//    @KafkaListener(topics = ComponentOutputLog.COMPONENT_EXPORT_QUEUE_TOPIC)
    public void consumerOutput(List<String> dataList) {
        for (String data : dataList) {
            if (isBlank(data)) {
                continue;
            }
            ComponentOutputLog outputLog = parseObject(data, ComponentOutputLog.class);
            if (Objects.isNull(outputLog)) {
                continue;
            }
            //保存输出记录
            saveExport(outputLog);
            //更新数据来源输出数据量
            saveOriginDataCount(outputLog);
        }
    }
}
